import asyncio
import json
import struct
import logging
from datetime import datetime, timedelta

# -------------------------- 协议常量（与客户端对齐）--------------------------
PING = 0x09  # 客户端PING心跳（单字节）
PONG = 0x0A  # 服务端PONG响应（单字节）
HEARTBEAT_TIMEOUT = 360  # 心跳超时时间（秒，与客户端360s一致）
MAX_FRAME_BYTES = 16 * 1024 * 1024  # 单帧最大长度（16MB，与客户端一致）
BYTE_ORDER = 'big'  # 帧头端序（大端，与客户端默认一致）
CHARSET = 'utf-8'  # 业务消息编码（UTF-8，与客户端一致）

# -------------------------- 日志配置 --------------------------
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger("ImageSocketServer")


# -------------------------- 业务消息类型（与客户端SocketMessage对齐）--------------------------
class MessageType:
    # 客户端发送给服务端的消息类型
    AI_REGISTER = 1  # TYPE_AI_REGISTER_FOR_DISPATCHER
    CLIENT_INSPECT_STATE = 2  # TYPE_CLIENT_INSPECT_STATE_FOR_DISPATCHER
    # 服务端发送给客户端的消息类型
    LINE_STATE_CHANGE = 101  # TYPE_LINE_STATE_CHANGE_FROM_DISPATCHER
    CAMERA_STATE_CHANGE = 102  # TYPE_CAMERA_STATE_CHANGE_FROM_DISPATCHER


# -------------------------- 服务端核心逻辑 --------------------------
async def read_exactly(reader: asyncio.StreamReader, n: int) -> bytes:
    """确保从流中读取指定长度的字节（处理asyncio.read可能未读满的情况）"""
    data = b''
    while len(data) < n:
        chunk = await reader.read(n - len(data))
        if not chunk:
            raise ConnectionResetError("客户端主动断开连接")
        data += chunk
    return data


async def heartbeat_monitor(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, last_recv_time: dict):
    """心跳超时监控：超过HEARTBEAT_TIMEOUT秒未收到数据则主动断开连接"""
    while True:
        await asyncio.sleep(10)  # 每10秒检查一次
        current_time = datetime.now()
        if (current_time - last_recv_time['time']).total_seconds() > HEARTBEAT_TIMEOUT:
            logger.warning(f"客户端 {writer.get_extra_info('peername')} 心跳超时（{HEARTBEAT_TIMEOUT}s），主动断开")
            writer.close()
            await writer.wait_closed()
            break


async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    """处理单个客户端连接：帧解析、心跳响应、业务消息处理"""
    peername = writer.get_extra_info('peername')
    logger.info(f"新客户端连接：{peername}")

    last_recv_time = {'time': datetime.now()}  # 用dict实现可变引用
    heartbeat_task = asyncio.create_task(heartbeat_monitor(reader, writer, last_recv_time))

    try:
        while True:
            # 1. 读取4字节帧头（长度）
            header = await read_exactly(reader, 4)
            payload_len = struct.unpack(f'>I', header)[0]  # >表示大端，I表示unsigned int（对应客户端int32）

            # 长度合法性检查
            if payload_len < 0 or payload_len > MAX_FRAME_BYTES:
                raise ValueError(f"非法帧长度：{payload_len}字节（超出16MB限制）")

            # 2. 读取Payload
            payload = await read_exactly(reader, payload_len)
            last_recv_time['time'] = datetime.now()  # 更新最后接收时间
            logger.debug(f"收到客户端 {peername} 数据：长度={payload_len}字节")

            # 3. 心跳消息处理
            if payload_len == 1:
                if payload[0] == PING:
                    logger.debug(f"收到客户端 {peername} PING心跳，回复PONG")
                    await send_frame(writer, bytes([PONG]))  # 回复PONG
                elif payload[0] == PONG:
                    logger.debug(f"收到客户端 {peername} PONG响应，忽略")
                continue  # 心跳消息不进入业务处理

            # 4. 业务消息处理（JSON解析）
            try:
                json_msg = payload.decode(CHARSET)
                msg = json.loads(json_msg)
                logger.info(f"收到客户端 {peername} 业务消息：{json_msg}")

                # 处理客户端注册消息

                await send_line_state_change(writer)

            except json.JSONDecodeError as e:
                logger.error(f"客户端 {peername} 消息JSON解析失败：{str(e)}，原始数据：{payload}")
            except KeyError as e:
                logger.error(f"客户端 {peername} 消息格式错误：缺少字段 {str(e)}")

    except ConnectionResetError:
        logger.info(f"客户端 {peername} 主动断开连接")
    except ValueError as e:
        logger.error(f"客户端 {peername} 数据非法：{str(e)}")
    except Exception as e:
        logger.error(f"客户端 {peername} 处理异常：{str(e)}", exc_info=True)
    finally:
        # 清理资源
        heartbeat_task.cancel()
        try:
            await heartbeat_task
        except asyncio.CancelledError:
            pass
        writer.close()
        await writer.wait_closed()
        logger.info(f"客户端 {peername} 连接已清理")


async def send_frame(writer: asyncio.StreamWriter, payload: bytes):
    """按协议发送帧：4字节长度前缀 + Payload"""
    payload_len = len(payload)
    header = struct.pack(f'>I', payload_len)  # 大端打包长度
    writer.write(header + payload)
    await writer.drain()  # 确保数据发送完成
    logger.debug(f"发送帧成功：长度={payload_len}字节")


async def send_line_state_change(writer: asyncio.StreamWriter):
    """发送产线状态变更示例消息（客户端可解析的业务消息）"""
    # line_state_msg = {"data":{"message":5,"datas":[{"prolineId":44,"state":0,"detail":{"左衣袖id1":"自动工作状态","右衣袖id2":"自动工作状态","衣身id3":"自动暂停状态","折叠id5":"自动工作状态"}}]},"module":20001}
    # line_state_msg = {"data":{"message":2,"datas":[{"prolineId":44,"offline":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17]},{"prolineId":43,"offline":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17]},{"prolineId":41,"offline":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17]}]},"module":20002}
    # line_state_msg = {"data":{"message":10,"date":1757298107000},"module":20003}

    line_state_msg = {"data":[{"byRule":"R1[0]","deviceAreaId":"35_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"34_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"22_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"24_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"35_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"19_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"22_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"25_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"33_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"35_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"34_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"22_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"24_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"35_2","proLineId":44,"type":"R1","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"19_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"22_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"25_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True},{"byRule":"R1[0]","deviceAreaId":"33_1","proLineId":43,"type":"R2","r1":{"n":5},"r2":{"k":3,"n":50},"r3":{"n":80,"percent":5},"triggered":True}],"module":20004}
    json_str = json.dumps(line_state_msg, ensure_ascii=False)
    await send_frame(writer, json_str.encode(CHARSET))
    logger.info(f"给客户端  发送产线状态变更消息：{json_str}")


async def start_server(host: str = '0.0.0.0', port: int = 9911):
    """启动TCP Socket服务端"""
    server = await asyncio.start_server(handle_client, host, port)
    addr = server.sockets[0].getsockname()
    logger.info(f"服务端已启动，监听地址：{addr}（TCP）")

    async with server:
        await server.serve_forever()


# -------------------------- 启动入口 --------------------------
if __name__ == "__main__":
    try:
        asyncio.run(start_server())
    except KeyboardInterrupt:
        logger.info("服务端收到中断信号，正在关闭...")
    except Exception as e:
        logger.error(f"服务端启动失败：{str(e)}", exc_info=True)